package com.leilei.source.mongo;

import com.alibaba.fastjson.JSON;
import com.mongodb.DBObject;
import com.mongodb.MongoClient;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Filters;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.bson.Document;
import org.bson.conversions.Bson;

import java.util.ArrayList;
import java.util.List;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/18 22:29
 * @desc
 */
public class MongoDBSource extends RichSourceFunction<User> {
    private MongoClient mongoClient;
    MongoDatabase mongoDatabase;
    MongoCollection<Document> userCollection;

    @Override
    public void open(Configuration parameters) throws Exception {
        ServerAddress serverAddress = new ServerAddress("xx", 27017);
        List<MongoCredential> credential = new ArrayList<>();
        //MongoCredential.createScramSha1Credential()三个参数分别为 用户名 数据库名称 密码
        MongoCredential mongoCredential = MongoCredential.createScramSha1Credential("xx", "xx", "xx..".toCharArray());
        credential.add(mongoCredential);
        mongoClient = new MongoClient(serverAddress, credential);
        mongoDatabase = mongoClient.getDatabase("test");
        userCollection = mongoDatabase.getCollection("user");
    }

    @Override
    public void run(SourceContext<User> ctx) throws Exception {
        Bson filter = Filters.eq("name", "张三");
        //指定查询过滤器查询
        FindIterable<Document> documents = userCollection.find(filter);
        MongoCursor<Document> iterator = documents.iterator();
        while (iterator.hasNext()) {
            Document next = iterator.next();
            System.out.println(JSON.parseObject(next.toJson(),User.class));
            //System.out.println(next);
        }

/*        List<Document> list = new ArrayList<>();
        for (int i = 1; i <= 3; i++) {
            Document document = new Document("name", "张三")
                    .append("_id", i)
                    .append("nick", "zhangsan")
                    .append("date", Integer.valueOf(LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE)));
            list.add(document);
        }
        userCollection.insertMany(list);*/
    }


    @Override
    public void cancel() {

    }

    @Override
    public void close() throws Exception {
        if (mongoClient != null) {
            mongoClient.close();
        }
    }
}
